package com.shujia.sink

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

object Demo5MysqlSink {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val studentDS: DataStream[String] = env.readTextFile("data/students.txt")


    //使用自定义sink
    studentDS.addSink(new MysqlSink)

    env.execute()
  }
}

/**
  *
  * SinkFunction
  * RichSinkFunction: 多个open和close方法
  */

class MysqlSink extends RichSinkFunction[String] {


  var con: Connection = _
  var stat: PreparedStatement = _

  /**
    * open: 在invoke 之前执行,每一个并行度中只执行一次
    *
    */
  override def open(parameters: Configuration): Unit = {
    println("open执行---")

    //加载驱动
    Class.forName("com.mysql.jdbc.Driver")
    //创建链接
    con = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=UTF-8", "root", "123456")
    //编写sql
    stat = con.prepareStatement("insert into students (id,name,age,gender,clazz) values(?,?,?,?,?)")

  }

  /**
    * close: 在invoke 之后执行,每一个并行度中只执行一次
    *
    */
  override def close(): Unit = {
    println("close执行---")
    stat.close()
    con.close()
  }

  /**
    * invoke: 每一条数据执行一次
    *
    * @param student :一条数据
    * @param context :上下文对象
    */
  override def invoke(student: String,
                      context: SinkFunction.Context): Unit = {
    val split: Array[String] = student.split(",")
    //按照位置指定数据
    stat.setLong(1, split(0).toLong)
    stat.setString(2, split(1))
    stat.setLong(3, split(2).toLong)
    stat.setString(4, split(3))
    stat.setString(5, split(4))
    //执行插入
    stat.execute()

  }
}
